package com.soso.base.google.mq;

import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;

import com.google.appengine.api.taskqueue.Queue;
import com.google.appengine.api.taskqueue.TaskOptions;
import com.google.appengine.api.taskqueue.TaskOptions.Method;
import com.soso.base.util.log.Logger;
import com.soso.base.util.other.SerialUtil;

public abstract class AbstractMessageQueue implements MessageQueue {
	
	public AbstractMessageQueue(){
		registerSubscribers();
	}
	@SuppressWarnings("rawtypes")
	protected Map<String, List<Class>> subscribers = new HashMap<String, List<Class>>();
	
	public void purge(Message message){
		if (message.getQueueName() == null) {
			Logger.error("QueueName is null in message " + message);
			return;
		}
		Queue queue = getQueue(getQueueName(message));
		queue.purge();
	}
	
	@Override
	public void publish(Message message) {
		if (message.getTopic() == null) {
			return;
		}
		if(Method.GET.name().equals(message.getMethod())){
			this.doGet(message);
		}else if(Method.PULL.name().equals(message.getMethod())){
			this.doPull(message);
		}else{
			this.doPost(message);
		}
	}
	
	
	private void doPost(Message message){
		if (message.getTopic() == null) {
			Logger.error("Topic is null in message " + message);
			return;
		}
		byte[] msgByte=SerialUtil.serialize(message);
		String TASK_CONTENT_TYPE = "application/x-java-serialized-object";
		Queue queue = getQueue(getQueueName(message));
		String url=MessageQueueServlet.MQ_URL+"/"+message.getTopic();
		Map<String,String> headers=message.getHeaders();
		TaskOptions taskOptions=withUrl(url).method(Method.POST);
		if(headers!=null && !headers.isEmpty()){
			taskOptions.headers(headers);
		}
        queue.add(taskOptions.payload(msgByte, TASK_CONTENT_TYPE ));
	}

	@SuppressWarnings("unchecked")
	private void doGet(Message message){
		Map<String, String> params=null;
		try{
			params =(Map<String, String>) message.getMessage();
		}catch(Exception e){
			this.doPost(message);
			return;
		}
		String queueName=getQueueName(message);
		Queue queue = getQueue(queueName);
		String url=message.getQueuePath();
		Map<String,String> headers=message.getHeaders();
		TaskOptions taskOptions=withUrl(url).method(Method.GET);
		if(headers!=null && !headers.isEmpty()){
			taskOptions.headers(headers);
		}
		//taskOptions.header("Host",BackendServiceFactory.getBackendService().getBackendAddress("backend1", 1));
		if (params == null) {
			return;
		}
		for(String key:params.keySet()){
			taskOptions.param(key, params.get(key));
		}
		queue.add(taskOptions);
	}
	
	private void doPull(Message message){
		if (message.getTopic() == null) {
			Logger.error("Topic is null in message " + message);
			return;
		}
		byte[] msgByte=SerialUtil.serialize(message);
		Queue queue = getQueue(getQueueName(message));
		Map<String,String> headers=message.getHeaders();
		TaskOptions taskOptions=TaskOptions.Builder.withMethod(Method.PULL);
		if(headers!=null && !headers.isEmpty()){
			taskOptions.headers(headers);
		}
        queue.add(taskOptions.payload(msgByte));
	}
	
	@Override
	public void execute(Message message) {
		if (subscribers.containsKey(message.getTopic())) {
			for (Class<?> subscriberClass : subscribers.get(message.getTopic())) {
				try {
					Subscriber subscriber = (Subscriber)subscriberClass.newInstance();
					subscriber.onMessage(message);
				}catch (IllegalAccessException e) {
					Logger.error(e.getMessage());
				}catch (InstantiationException e) {
					Logger.error(e.getMessage());
				}
			}
		}
		if (!StringUtils.isEmpty(message.getCommandClassName())) {
			try {
				Class<?> clazz = loadClass(message.getCommandClassName());
				if (clazz != null) {
					Subscriber subscriber = (Subscriber)clazz.newInstance();
					subscriber.onMessage(message);
				}
				else {
					Logger.error("Command class not found: " + message.getCommandClassName());
				}
			}
			catch (IllegalAccessException e) {
				Logger.error(e.getMessage());
			}
			catch (InstantiationException e) {
				Logger.error(e.getMessage());
			}
		}
	}
	
	@Override
	public void subscribe(Enum<?> topic, Class<?> subscriber){
		subscribe(topic.name(), subscriber);
	}
	
	@SuppressWarnings("rawtypes")
	@Override
	public void subscribe(String topic, Class<?> subscriber) {
		if (!subscribers.containsKey(topic)) {
			subscribers.put(topic, new ArrayList<Class>());
		}
		if (!subscribers.get(topic).contains(subscriber)) {
			subscribers.get(topic).add(subscriber);
		}
	}

	@Override
	public void unsubscribe(String topic, Class<?> subscriber) {
		if (subscribers.containsKey(topic)) {
			if (subscribers.get(topic).contains(subscriber)) {
				subscribers.get(topic).remove(subscriber);
			}
		}
	}
	private Class<?> loadClass(String name) {
		try {
			return this.getClass().getClassLoader().loadClass(name);
		}catch (ClassNotFoundException e) {
			return null;
		}
	}
	protected String getQueueName(Message message) {
		if (message.getQueueName() == null) {
			return "default";
		}
		return message.getQueueName().name();
	}
	protected abstract void registerSubscribers() ;
	protected abstract Queue getQueue(String queueName);
}
